package edu.buaa.energy.client;

import com.google.common.util.concurrent.ListenableFuture;
import edu.buaa.common.client.MariaDbExecutorClient;
import edu.buaa.common.transaction.AbstractTransaction;
import edu.buaa.energy.EnergyDataTxGenerator;
import edu.buaa.energy.transaction.*;
import edu.buaa.utils.Pair;
import edu.buaa.utils.Triple;

import java.sql.*;
import java.util.*;
import java.util.concurrent.*;

/**
 * Created by sjh. 2021.11.18
 */
public class MariaDBClient extends MariaDbExecutorClient {

    public MariaDBClient(String serverHost, int parallelCnt) throws Exception {
        super(serverHost, "energy", parallelCnt);
    }

    @Override
    protected List<String> createTables() {
        return Arrays.asList(
                "CREATE table energy_node(raw_id int PRIMARY KEY, name varchar(255), country varchar(255), volt int, lat float, lon float)",
                "CREATE table electric_line(r_start int, r_end int, x float, y float, r_limit int, r_length float)",
                "CREATE table temporal_status(node_id int, st_time TIMESTAMP(0), en_time TIMESTAMP(0), " + //note: st is closed [, en is not closed ) because mariadb's restrict.
                        "load_signal FLOAT, solar_cosmo_p FLOAT, solar_cosmo_u FLOAT, solar_ecmwf_p FLOAT, solar_ecmwf_u FLOAT, wind_cosmo_p FLOAT, wind_cosmo_u FLOAT, wind_ecmwf_p FLOAT, wind_ecmwf_u FLOAT, " +
                        "PERIOD FOR time_period(st_time, en_time), PRIMARY KEY (node_id,st_time))",
                // InnoDB默认对主键建立聚簇索引。如果不指定主键，InnoDB会用一个具有唯一且非空值的索引来代替。如果不存在这样的索引，InnoDB会定义一个隐藏的主键，然后对其建立聚簇索引。
                // stmt.execute("create index ind_rid_st on temporal_status(node_id, st_time)");
                "create index ind_st_en on temporal_status(st_time, en_time)"
        );
    }

    @Override
    public ListenableFuture<ServerResponse> execute(AbstractTransaction tx) throws InterruptedException {
        switch (tx.getTxType()) {
            case tx_import_static_data:
                return this.submit(execute((ImportStaticDataTx) tx));
            case tx_import_temporal_data:
                return this.submit(execute((ImportTemporalDataTx) tx), tx.getSection());
            case tx_query_snapshot:
                return this.submit(execute((SnapshotQueryTx) tx));
            case tx_update_temporal_data:
                return this.submit(execute((UpdateTemporalDataTx) tx));
            case tx_query_snapshot_aggr_max:
                return this.submit(execute((SnapshotAggrMaxTx) tx));
            case tx_query_snapshot_aggr_duration:
                return this.submit(execute((SnapshotAggrDurationTx) tx));
            case tx_query_entity_history:
                return this.submit(execute((EntityHistoryTx) tx));
            case tx_query_road_by_temporal_condition:
                return this.submit(execute((EntityTemporalConditionTx) tx));
            default:
                throw new UnsupportedOperationException();
        }
    }

    private Req execute(ImportStaticDataTx tx) {
        return conn -> {
            try (PreparedStatement stat1 = conn.prepareStatement("INSERT INTO energy_node VALUES (?,?,?,?,?,?)");
                 PreparedStatement stat2 = conn.prepareStatement("INSERT INTO electric_line VALUES (?,?,?,?,?,?)")) {
                conn.setAutoCommit(false);
                for (ImportStaticDataTx.EnergyNode p : tx.getNodes()) {
                    stat1.setInt(1, p.getRawId());
                    stat1.setString(2, p.getName());
                    stat1.setString(3, p.getCountry());
                    stat1.setInt(4, p.getVoltage());
                    stat1.setFloat(5, p.getLatitude());
                    stat1.setFloat(6, p.getLongitude());
                    stat1.addBatch();
                }
                stat1.executeBatch();
                for (ImportStaticDataTx.EnergyRel r : tx.getRels()) {
                    stat2.setInt(1, r.getFromRawId());
                    stat2.setInt(2, r.getToRawId());
                    stat2.setFloat(3, r.getX());
                    stat2.setFloat(4, r.getY());
                    stat2.setInt(5, r.getLimit());
                    stat2.setFloat(6, r.getLength());
                    stat2.addBatch();
                }
                stat2.executeBatch();
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            return new AbstractTransaction.Result();
        };
    }

    public Req execute(ImportTemporalDataTx tx) {
        return conn -> {
            try (PreparedStatement stmt0 = conn.prepareStatement("insert into temporal_status(node_id,st_time,en_time,"+
                    String.join(",", EnergyDataTxGenerator.properties)+") values(?,?,?,?,?,?,?,?,?,?,?,?)")) {
                conn.setAutoCommit(false);
                for (ImportTemporalDataTx.EnergyStatus s : tx.data) {
                    stmt0.setInt(1, s.getNodeId());
                    stmt0.setTimestamp(2, new Timestamp(s.getTime() * 1000L));
                    stmt0.setTimestamp(3, new Timestamp((s.getTime() + 3600) * 1000L));
                    List<Float> vals = s.getStatus();
                    for (int i = 4; i < 4 + vals.size(); i++) {
                        stmt0.setFloat(i, vals.get(i - 4));
                    }
                    stmt0.addBatch();
                }
                stmt0.executeBatch();
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                conn.setAutoCommit(true);
            }
            return new AbstractTransaction.Result();
        };
    }

    private Req execute(UpdateTemporalDataTx tx) {
        return conn -> {
            conn.setAutoCommit(true);
            StringBuilder sql = new StringBuilder();
            sql.append("UPDATE temporal_status FOR PORTION OF time_period FROM '")
                    .append(new Timestamp(tx.getStartTime() * 1000L)).append("' TO '")
                    .append(new Timestamp((tx.getEndTime()+1) * 1000L)).append("' SET ");
            List<Float> vals = tx.getValues();
            for(int i = 0; i< EnergyDataTxGenerator.properties.length; i++) {
                String propName = EnergyDataTxGenerator.properties[i];
                Float val = vals.get(i);
                sql.append(propName).append("=").append(val).append(',');
            }
            sql.setLength(sql.length()-1);
            sql.append(" WHERE node_id=").append(tx.getEntityId());
            System.out.println(sql);
            try (PreparedStatement stmt0 = conn.prepareStatement(sql.toString())) {
                stmt0.executeQuery();
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return new AbstractTransaction.Result();
        };
    }

    private Req execute(SnapshotQueryTx tx) {
        return conn -> {
            List<Pair<Integer, Float>> res = new ArrayList<>();
            String sql = "select node_id," + tx.getPropertyName() + " as property, st_time, en_time from temporal_status as ts " +
                    "where  \"" + new Timestamp((long) tx.getTimestamp() * 1000L) + "\" between st_time and en_time group by node_id";
            System.out.println(sql);
            try (PreparedStatement stat = conn.prepareStatement(sql)) {
                ResultSet rs = stat.executeQuery();
                while (rs.next()) {
                    if(rs.getTimestamp("en_time").getTime() > tx.getTimestamp()*1000L) {
                        res.add(Pair.of(rs.getInt("node_id"), rs.getFloat("property")));
                    }
                }
            }
            SnapshotQueryTx.Result result = new SnapshotQueryTx.Result();
            result.setStatus(res);
            return result;
        };
    }


    private Req execute(SnapshotAggrMaxTx tx) {
        return conn -> {
            List<Pair<Integer, Float>> res = new ArrayList<>();
            String sql = "select node_id, max(" + tx.getP() + ") as max_p from temporal_status " +
                    "WHERE st_time<=\"" + new Timestamp((long) tx.getT1() * 1000L) + "\" and en_time > \"" + new Timestamp((long) tx.getT0() * 1000L) + "\" group by node_id";
            System.out.println(sql);
            try (PreparedStatement stat = conn.prepareStatement(sql)) {
                conn.setAutoCommit(false);
                ResultSet rs = stat.executeQuery();
                while(rs.next()) {
                    res.add(Pair.of(rs.getInt("node_id"), rs.getFloat("max_p")));
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            SnapshotAggrMaxTx.Result result = new SnapshotAggrMaxTx.Result();
            result.setPropMaxValue(res);
            return result;
        };
    }

    private Req execute(SnapshotAggrDurationTx tx) {
        return conn -> {
            List<Triple<Integer, Float, Integer>> res = new ArrayList<>();
            int t0 = tx.getT0(), t1 = tx.getT1();
            String sql = "select node_id, " + tx.getP() + " as property, st_time, en_time from temporal_status where st_time <= \"" +
                    new Timestamp((long) t1 * 1000L) + "\" and en_time > \"" +
                    new Timestamp((long) t0 * 1000L) + "\" order by node_id, st_time";
            System.out.println(sql);
            TreeSet<Float> intervalStarts = tx.getIntStartTreeSet();
            try (PreparedStatement stat = conn.prepareStatement(sql);
                 ResultSet rs = stat.executeQuery()) {
                int lastId = -1;
                Map<Float, Integer> map = new HashMap<>();
                while (rs.next()) {
                    int id = rs.getInt("node_id");
                    if(lastId!=id){
                        if(lastId!=-1){
                            for(Map.Entry<Float, Integer> e : map.entrySet()){
                                res.add(Triple.of(lastId, e.getKey(), e.getValue()));
                            }
                        }
                        map.clear();
                    }
                    float value = rs.getFloat("property");
                    int st = (int) (rs.getTimestamp("st_time").getTime() / 1000L);
                    int en = (int) (rs.getTimestamp("en_time").getTime() / 1000L) - 1;
                    int left = Math.max(t0, st);
                    int right = Math.min(t1, en);
                    int duration = right - left + 1;
                    Float valGrp = intervalStarts.floor(value);
                    if (valGrp != null) {
                        map.merge(valGrp, duration, Integer::sum);
                    }
                    lastId = id;
                }
                if(lastId!=-1){
                    for(Map.Entry<Float, Integer> e : map.entrySet()){
                        res.add(Triple.of(lastId, e.getKey(), e.getValue()));
                    }
                }
            }
            SnapshotAggrDurationTx.Result result = new SnapshotAggrDurationTx.Result();
            result.setStatusDuration(res);
            return result;
        };
    }

    private Req execute(EntityHistoryTx tx) {
        return conn -> {
            List<Triple<Integer, Integer, Float>> res = new ArrayList<>();
            int t0 = tx.getBeginTime(), t1 = tx.getEndTime();
            String sql = "select " + tx.getProp() + " as property, st_time, en_time from temporal_status where st_time <= \"" +
                    new Timestamp((long) t1 * 1000L) + "\" and en_time > \"" +
                    new Timestamp((long) t0 * 1000L) + "\" and node_id="+tx.getEntity()+" order by st_time";
            System.out.println(sql);
            try (PreparedStatement stat = conn.prepareStatement(sql);
                 ResultSet rs = stat.executeQuery()) {
                while (rs.next()) {
                    float value = rs.getFloat("property");
                    int st = (int) (rs.getTimestamp("st_time").getTime() / 1000L);
                    int en = (int) (rs.getTimestamp("en_time").getTime() / 1000L) - 1;
                    int beginT = Math.max(st, tx.getBeginTime());
                    int endT = Math.min(en, tx.getEndTime());
                    if(beginT<=endT) res.add(Triple.of(beginT, endT, value));
                }
            }
            EntityHistoryTx.Result result = new EntityHistoryTx.Result();
            result.setHistory(res);
            return result;
        };
    }

    private Req execute(EntityTemporalConditionTx tx) {
        return conn -> {
            List<Integer> res = new ArrayList<>();
            String sql = "select distinct node_id from temporal_status where st_time <= \"" + new Timestamp((long) tx.getT1() * 1000L) +
                    "\" and en_time > \"" + new Timestamp((long) tx.getT0() * 1000L) + "\" and " + tx.getP() +
                    " between " + tx.getVMin() + " and " + tx.getVMax() +" group by node_id";
            System.out.println(sql);
            try (PreparedStatement stat = conn.prepareStatement(sql)) {
                ResultSet rs = stat.executeQuery();
                while (rs.next()) {
                    res.add(rs.getInt("node_id"));
                }
            }
            EntityTemporalConditionTx.Result result = new EntityTemporalConditionTx.Result();
            result.setEntities(res);
            return result;
        };
    }
}

